package SQL

import org.apache.spark.sql
import org.apache.spark.sql.catalyst.encoders.RowEncoder
import org.apache.spark.sql.{DataFrame, Dataset, Row}
import org.junit.Test

class Dataset {
  val spark = new sql.SparkSession.Builder()
    .master("local[6]")
    .appName("DataSet")
    .getOrCreate()
  //导入隐式转换
  import spark.implicits._

  @Test
  def DatasetTest1(): Unit = {
    //创建RDD
    val sourceRDD = spark.sparkContext.parallelize(Seq(person("zhangsan", 10), person("lisi", 10), person("wangwu", 14), person("zhaoliu", 11)))
    //转换成Dataset
    val dataset = sourceRDD.toDS()
    //找到年龄大于10岁的
    //DataSet支持RDD（强类型）的API
    dataset
      .filter(item => item.age > 10)
      .show()
    //Dataset也支持弱类型的API
    dataset
      .filter('age > 10)
      .show()
    //可以直接编写SQL表达式
    dataset
      .where("age > 10")
      .show()
  }
  @Test
  def DataFrameTest1(): Unit ={
    //创建DataFrame
    val dataframe = spark.sparkContext.parallelize(Seq(person("zhangsan", 10), person("lisi", 10), person("wangwu", 14), person("zhaoliu", 11))).toDF()
    //查询
    dataframe.where('age > 10)
      .select('name)
      .show()
  }
  @Test
  def DataFrameTest2(): Unit ={
    val rdd = spark.sparkContext.parallelize(Seq(person("zhangsan", 10), person("lisi", 10), person("wangwu", 14), person("zhaoliu", 11)))
    //创建DataFrame的三种方式
    //1、toDF
    val df1: DataFrame = rdd.toDF()
    //2、createDataFrame
    val df2 = spark.createDataFrame(rdd)
    //3、read
    val rdd3 = spark.read.csv("resource/data.csv")
  }
  @Test
  def DataFrameTest3(): Unit ={
    val sourceDF = spark.read
      //第一行为列名
      .option("header", value = true)
      .csv("C:\\Users\\HR\\Desktop\\A.csv")

    //    sourceDF.show(10)
    //查看结构信息
    //    sourceDF.printSchema()
    //处理
    sourceDF.select('Year, 'month, 'PM_Dongsi)
      .where('PM_Dongsi =!= "NA")
      .groupBy('Year, 'month)
      .count()
      .show()
    //直接使用SQL
    //注册为临时表
    sourceDF.createOrReplaceTempView("pm")
    val resultDF = spark.sql("select * from pm")
    spark.stop()
  }
  @Test
  def set_and_frame(): Unit ={
    val personlist = Seq(person("张三", 10), person("李四", 19))
    //DataFrame是弱类型,操作的是row对象
    val df: DataFrame = personlist.toDF()
    df.map((row:Row)=>Row(row.get(0),row.getAs[Int](1)*2))(RowEncoder.apply(df.schema)).show()
    //Dataset是强类型，操作的是存储的对象
    val ds: sql.Dataset[person] = personlist.toDS()
    ds.map(person=>(person.name,person.age*2)).show()
  }
}

case class person(name: String, age: Int)